package com.dongdongshop.mq;

import com.alibaba.fastjson.JSONObject;
import com.aliyun.dysmsapi20170525.Client;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Component
@RocketMQMessageListener(consumerGroup = "dk2207-consumer", messageModel = MessageModel.CLUSTERING, topic = "dk2207-topic")
public class RocketMqConsumer implements RocketMQListener<String>, RocketMQPushConsumerLifecycleListener {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private Client client;
    @Value("${dysms.signName}")
    private String signName;
    @Value("${dysms.templateCode}")
    private String templateCode;

    @Override
    public void onMessage(String message) {

    }

    @Override
    public void prepareStart(DefaultMQPushConsumer consumer) {
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
                                                            ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                //获取消息内容
                MessageExt messageExt = list.get(0);
                byte[] body = messageExt.getBody();//消息内容
                String message = new String(body);
                Map map = JSONObject.parseObject(message, Map.class);
                //取出内容手机号
                String phoneNum = map.get("phoneNum").toString();
                //消息id
                String msgId = messageExt.getMsgId();
                //获取主题
                String topic = messageExt.getTopic();
                //重试次数
                int reconsumeTimes = messageExt.getReconsumeTimes();
                try {
                    //消息幂等性问题,判断是否存在消息+id
                    Boolean hasKey = redisTemplate.hasKey("MESSAGE" + msgId);
                    //存在,返回消费成功,不存在,则去消费
                    if (!hasKey) {
                        //随机生成数字
                        String code = RandomStringUtils.randomNumeric(6);
                        JSONObject json = new JSONObject();
                        json.put("sms_code", code);
                        com.aliyun.dysmsapi20170525.models.SendSmsRequest sendSmsRequest = new com.aliyun.dysmsapi20170525.models.SendSmsRequest()
                                .setPhoneNumbers(phoneNum)
                                .setSignName(signName)
                                .setTemplateCode(templateCode)
                                .setTemplateParam(json.toJSONString());
                        com.aliyun.teautil.models.RuntimeOptions runtime = new com.aliyun.teautil.models.RuntimeOptions();
                        // 复制代码运行请自行打印 API 的返回值
                        client.sendSmsWithOptions(sendSmsRequest, runtime);
                        //消费成功,存入redis中
                        redisTemplate.boundValueOps("MESSAGE" + msgId).set("消费成功");
                        //在存入redis中,将验证码 与 手机号存入,并设置过期时间为30分钟
                        redisTemplate.boundHashOps("REGISTER_CODE").put(phoneNum,code);
                        //设置过期时间1800秒
                        redisTemplate.boundHashOps("REGISTER_CODE").getOperations().expire(phoneNum,1800, TimeUnit.SECONDS);

                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                } catch (Exception e) {
                    e.printStackTrace();
                    //消息重试次数达到2次,写入私信消息表中,返回消息消费成功
                    if (reconsumeTimes >= 1) {
                       /* TDeadqueue tDeadqueue = new TDeadqueue();
                        tDeadqueue.setBody(message);
                        tDeadqueue.setTopic(topic);
                        tDeadqueue.setMessageId(msgId);
                        tDeadqueue.setFlag(0);
                        tDeadqueue.setError(e.getMessage());
                        tDeadqueue.setTime(new Date());
                        tDeadqueueMapper.insertSelective(tDeadqueue);*/
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    }
                    //未达到两次去重试消息
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
            }
        });
    }
}
